package com.neusoft.reactor;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.util.Random;

/**
 * @author lichunlong lcl253137@alibaba-inc.com
 * @desc Reactor框架的Flux集合类示例
 * @date 2023-01-17
 */
public class FluxDemo {

    public static void main(String[] args) throws InterruptedException {
        test5();

        // 配合test5多线程方法使用
        Thread.currentThread().join(1000L);
    }

    /**
     * Flux第一个应用
     */
    public static void test1() {
        Flux<Integer> flux = Flux.just(1, 2, 3);
        // 订阅并处理
        flux.subscribe(val -> System.out.printf("Thread[ %s ]: %d \n", Thread.currentThread().getName(), val));
    }

    /**
     * Flux流式写法，test1方法的简化版
     */
    public static void test2() {
        // 订阅处理 - 流式编程
        Flux.just(1, 2, 3).subscribe(val -> System.out.printf("Thread[ %s ]: %d \n", Thread.currentThread().getName(), val));
    }

    /**
     * 多方法处理: onNext、onError、onSubscribe
     */
    public static void test3() {
        Random random = new Random();
        Flux.just(1, 2, 3).map(val -> {
            if (random.nextInt(4) == 3) {
                throw new RuntimeException("exception msg");
            }
            return val + 1;
        }).subscribe(FluxDemo::println, // 处理数据 onNext()方法
                FluxDemo::println, // 处理数据 onError()方法
                () -> FluxDemo.println("subscription is completed")
        );
    }

    /**
     * 迭代器处理概念
     */
    public static void test4() {
        Flux.generate(() -> 0, (value, sink) -> {
            if (value == 3) {
                // 主动完成，终止生成方法
                sink.complete();
            } else {
                sink.next("value=" + value);
            }
            return value + 1;
        }).subscribe(FluxDemo::println);
    }

    /**
     * 多线程运行，上述都是main主线程运行
     */
    public static void test5() {
        // main主线程运行
        Flux.range(1, 10).subscribeOn(Schedulers.immediate()).subscribe(FluxDemo::println);

        // 单线程运行，非main主线程
        Flux.range(1, 10).subscribeOn(Schedulers.single()).subscribe(FluxDemo::println);

        // 多线程运行，弹性增加线程
        Flux.range(1, 10).subscribeOn(Schedulers.boundedElastic()).subscribe(FluxDemo::println);

        // 多线程并行运行
        Flux.range(1, 10).subscribeOn(Schedulers.parallel()).subscribe(FluxDemo::println);
    }

    public static void println(Object msg) {
        System.out.printf("Thread[ %s ]: %s \n", Thread.currentThread().getName(), msg);
    }
}
